package com.shujia.source

import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet}

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.source.{ParallelSourceFunction, RichSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala._

object Demo6MysqlSource {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(2)
    //使用mysqlSource
    val studentDS: DataStream[String] = env.addSource(new MysqlSource)

    studentDS.print()

    env.execute()

  }

}

/**
  * 有了flink的自定义source, 所有的数据源只要提供了java api 都可以使用flink来读取
  *
  * SourceFunction: 单并行度的source
  * ParallelSourceFunction: 多并行度的source
  * RichSourceFunction: 多了open和close方法
  * RichParallelSourceFunction:多了open和close方法
  */

class MysqlSource extends RichSourceFunction[String] {

  var con: Connection = _
  var stat: PreparedStatement = _

  /**
    * open: 在run方法之前执行,一般用于创建链接
    *
    * @param parameters : 配置文件对象
    */
  override def open(parameters: Configuration): Unit = {
    //1加载驱动
    Class.forName("com.mysql.jdbc.Driver")
    //创建链接
    con = DriverManager.getConnection("jdbc:mysql://master:3306/bigdata", "root", "123456")
    //执行sql
    stat = con.prepareStatement("select * from students")
  }

  /**
    * close在run 方法之后执行,一i版用于关闭链接
    *
    */
  override def close(): Unit = {
    //关闭链接
    stat.close()
    con.close()

  }


  /**
    * run :用于读取数据的方法,在单并行度的情况下只执行一次
    *
    * @param ctx :用于将数据发生到下游
    */
  override def run(ctx: SourceFunction.SourceContext[String]): Unit = {
    /**
      * 使用jdbc读取mysql中的数据,将数据发生到下游
      */

    //执行查询
    val resultSet: ResultSet = stat.executeQuery()
    //解析数据
    while (resultSet.next()) {
      val id: Long = resultSet.getLong("id")
      val name: String = resultSet.getString("name")
      val age: Long = resultSet.getLong("age")
      val gender: String = resultSet.getString("gender")
      val clazz: String = resultSet.getString("clazz")

      //将数据发生到下游
      ctx.collect(s"$id\t$name\t$age\t$gender\t$clazz")
    }

  }


  //cancel方法会再job被取消的时候执行,一般用于回收资源
  override def cancel(): Unit = {}

}
